package net.bwie.realtime.mall.dim.funcation;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;

/**
 * @ClassName：DimHBaseMapFunction
 * @Author: ZddddQ
 * @Date: 2025/6/3 13:39
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class DimHBaseMapFunction extends RichMapFunction<String, String> {

    // 维度表字段映射集合
    private HashMap<String, String> hashMap;

    // 构造函数
    public DimHBaseMapFunction(HashMap<String, String> hashMap) {
        this.hashMap = hashMap;
    }

    //  初始化
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public String map(String value) throws Exception {

        //put 'dim_base_dic', '1001', 'info:dic_code', '10'
        //表名称、rowKey值（主键字段名称）、列簇名称、列名称、列对应值

        // 解析JSON
        JSONObject jsonObject = JSON.parseObject(value);
        // 添加HBASE表名称
        jsonObject.put("hbase_table_name", "dim_" + jsonObject.getString("table_name"));
        // 添加列簇名称
        jsonObject.put("family_name", "info");
        // 添加rowKey值对应的名称
        jsonObject.put("row_key_column", hashMap.get(jsonObject.getString("table_name")));

        // 返回
        return jsonObject.toJSONString();
    }

    // 释放资源
    @Override
    public void close() throws Exception {
        super.close();
    }

}
